package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.function.DimAsyncFunction;
import com.atguigu.gmall.realtime.beans.DimJoinFunction;
import com.atguigu.gmall.realtime.beans.InteractionReviewBean;
import com.atguigu.gmall.realtime.utils.DateFormatUtil;
import com.atguigu.gmall.realtime.utils.MyClickhouseUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

public class DwsInteractionReviewWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        String topic = "dwd_interaction_review";
        String groupId = "dws_interaction_review_group";
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaDS = env.addSource(kafkaConsumer);
        SingleOutputStreamOperator<JSONObject> jsonDS = kafkaDS.map(JSON::parseObject);
        SingleOutputStreamOperator<JSONObject> withWatermarkDS = jsonDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps()
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                    @Override
                    public long extractTimestamp(JSONObject element, long recordTimestamp) {
                        return element.getLong("ts") * 1000;
                    }
                }));

        SingleOutputStreamOperator<InteractionReviewBean> beanDS = withWatermarkDS.map(new MapFunction<JSONObject, InteractionReviewBean>() {
            @Override
            public InteractionReviewBean map(JSONObject value) throws Exception {
                Long totalGRUser = 0L;
                Long totalStars = Long.valueOf(value.getString("review_stars"));
                String courseId = value.getString("course_id");
                Long ts = value.getLong("ts");
                if (totalStars >= 4) {
                    totalGRUser = 1L;
                }
                InteractionReviewBean bean = InteractionReviewBean.builder()
                        .courseId(courseId)
                        .totalUser(1L)
                        .totalGRUser(totalGRUser)
                        .totalStars(totalStars)
                        .ts(ts)
                        .build();

                return bean;
            }
        });
        KeyedStream<InteractionReviewBean, String> keyedDS = beanDS.keyBy(InteractionReviewBean::getCourseId);

        WindowedStream<InteractionReviewBean, String, TimeWindow> windowDS = keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        SingleOutputStreamOperator<InteractionReviewBean> reduceDS = windowDS.reduce(new ReduceFunction<InteractionReviewBean>() {
            @Override
            public InteractionReviewBean reduce(InteractionReviewBean value1, InteractionReviewBean value2) throws Exception {
                value1.setTotalGRUser(value1.getTotalGRUser() + value2.getTotalGRUser());
                value1.setTotalUser(value1.getTotalUser() + value2.getTotalUser());
                value1.setTotalStars(value1.getTotalStars() + value2.getTotalStars());
                return value1;
            }
        }, new WindowFunction<InteractionReviewBean, InteractionReviewBean, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow window, Iterable<InteractionReviewBean> input, Collector<InteractionReviewBean> out) throws Exception {
                String stt = DateFormatUtil.toYmdHms(window.getStart());
                String edt = DateFormatUtil.toYmdHms(window.getEnd());
                for (InteractionReviewBean value : input) {
                    value.setStt(stt);
                    value.setEdt(edt);
                    value.setTs(System.currentTimeMillis());
                    out.collect(value);
                }
            }
        });
/*
        SingleOutputStreamOperator<InteractionReviewBean> reduceDS = windowDS.reduce(new ReduceFunction<InteractionReviewBean>() {
            @Override
            public InteractionReviewBean reduce(InteractionReviewBean value1, InteractionReviewBean value2) throws Exception {
                value1.setTotalGRUser(value1.getTotalGRUser() + value2.getTotalGRUser());
                value1.setTotalUser(value1.getTotalUser() + value2.getTotalUser());
                value1.setTotalStars(value1.getTotalStars() + value2.getTotalStars());
                return value1;
            }
        }, new AllWindowFunction<InteractionReviewBean, InteractionReviewBean, TimeWindow>() {
            @Override
            public void apply(TimeWindow window, Iterable<InteractionReviewBean> values, Collector<InteractionReviewBean> out) throws Exception {
                String stt = DateFormatUtil.toYmdHms(window.getStart());
                String edt = DateFormatUtil.toYmdHms(window.getEnd());
                for (InteractionReviewBean value : values) {
                    value.setStt(stt);
                    value.setEdt(edt);
                    value.setTs(System.currentTimeMillis());
                    out.collect(value);
                }
            }
        });*/
        SingleOutputStreamOperator<InteractionReviewBean> withNameDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<InteractionReviewBean>("dim_course_info") {
                    @Override
                    public void join(InteractionReviewBean obj, JSONObject dimInfoJsonObj) {

                        obj.setCourseName(dimInfoJsonObj.getString("COURSE_NAME"));
                    }

                    @Override
                    public String getKey(InteractionReviewBean obj) {
                        return obj.getCourseId();
                    }
                },
                60, TimeUnit.SECONDS);
        withNameDS.print();
        withNameDS.addSink(
                MyClickhouseUtil.getSinkFunction("insert into dws_interaction_review_window values(?,?,?,?,?,?,?,?)"));
        env.execute();
    }
}

